package com.dmp.storm.bolt;

import org.apache.commons.lang3.StringUtils;

import java.util.HashMap;
import java.util.Map;

import static com.dmp.storm.base.BaseMapKey.MAPKEY_RESULTMAP_BOLTMSG;
import static com.dmp.util.Constant.DATA_SEPARATOR_DEFAULT;
import static com.dmp.util.Constant.FIELD_SEPARATOR_DEFAULT;

/**
 * 分词匹配bolt
 * 
 * @author guozh3
 *
 */
public class WordSplitBolt extends BaseMapBolt {
	private static final long serialVersionUID = 1L;
	public static String[] fieldNames;
	private String separator = DATA_SEPARATOR_DEFAULT;
	private String fields;
	private String fieldSeparator = FIELD_SEPARATOR_DEFAULT;
	
	@Override
	public void prepare() {
		if(StringUtils.isNotBlank(fields)){
			fieldNames = StringUtils.split(fields, fieldSeparator);
		}else{
			System.err.println("field is null");
			System.exit(-1);
		}
	}
	
	// 将从kafka取到的消息按照“,”分词，并将数据与输入字段进行匹配
	@Override
	public Map<String, Object> process(Map<String, Object> map) {
		Map<String, Object> resultMap = null;
		
		if(fieldNames != null && fieldNames.length > 0){
			String record = (String) map.get(MAPKEY_RESULTMAP_BOLTMSG);
			//System.out.println(record);
			if(StringUtils.isNotBlank(record)){
				String[] values = record.split(this.separator);
				resultMap = buildFieldValue(fieldNames, values);
			}
		}
		 System.out.println("WordSplitBolt1");
		return resultMap;
	}
	
	private Map<String, Object> buildFieldValue(String[] fields, String[] values){
		Map<String, Object> record = new HashMap<String, Object>();
		for(int i=0; i<fields.length; i++){
			if(i < values.length){
				record.put(fields[i], values[i]);
			} else {
				record.put(fields[i], null);
			}
		}
		
		return record;
	}

	public void setSeparator(String separator) {
		this.separator = separator;
	}

	public void setFields(String fields) {
		this.fields = fields;
	}
	
}
